import threading
from kafka import KafkaProducer
import time

broker_list = "10.0.88.277:9092,10.0.88.228:9092,10.0.11.204:9092"
topic_name = "api-gateway-test"
num_messages = 50

# 读取 data.txt 文件内容
with open('data.txt', 'r', encoding='utf-8') as file:
    data_content = file.read()

# 定义一个函数来发送消息
def produce_messages(start_partition, end_partition, message_data):
    producer = KafkaProducer(bootstrap_servers=broker_list)

    # Produce messages
    for i in range(num_messages):
        for partition in range(start_partition, end_partition + 1):
            message = f'Message-{i} from partition-{partition}: {message_data}'.encode('utf-8')
            producer.send(topic=topic_name, value=message, partition=partition)

    producer.flush()
    producer.close()

# 线程列表
threads = []

# 定义线程数量
num_threads = 50  # 修改为正确的线程数

# 计算每个线程需要处理的分区数量
partitions_per_thread = 50 // num_threads

for i in range(num_threads):
    start_partition = i * partitions_per_thread
    # 确保最后一个线程能够处理剩下的所有分区
    end_partition = start_partition + partitions_per_thread - 1 if i < num_threads - 1 else 49

    thread = threading.Thread(target=produce_messages, args=(start_partition, end_partition, data_content))
    threads.append(thread)
    thread.start()

# 等待所有线程完成
for thread in threads:
    thread.join()

print("所有消息已发送完毕。")